#!/usr/bin/env python

#   
#     t_newsbiff.py - Test script for newsbiff
#
#  Copyright (c) 2007, Alex Marandon
#  All rights reserved.
#
#  Redistribution and use in source and binary forms, with or without
#  modification, are permitted provided that the following conditions are met:
#  
#    * Redistributions of source code must retain the above copyright notice,
#      this list of conditions and the following disclaimer.
#    * Redistributions in binary form must reproduce the above copyright notice,
#      this list of conditions and the following disclaimer in the documentation
#      and/or other materials provided with the distribution.
#    * Neither the name of newsbiff nor the names of its contributors
#      may be used to endorse or promote products derived from this software 
#      without specific prior written permission.
#  
#  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
#  AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
#  IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
#  ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
#  LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
#  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
#  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
#  INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
#  CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
#  ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
#  POSSIBILITY OF SUCH DAMAGE.

"""
See README file for directions on how to set the environment required to run
this script.
"""

import sys, os, signal, unittest, subprocess, time, nntplib, shutil
sys.path.append('..')
from newsbiff import Conf, ConfigError, SearchEngine, AtomRenderer, RssRenderer
import elementtree.ElementTree as ET

SNROOT=os.path.expanduser('~/newsbiff/tests/spool')
SNHOST='localhost'
SNPORT=1190
DELAY=0.2
GROUPS=('fr.petites-annonces.musique', 'fr.emplois.offres', 'fr.test',
        'fr.comp.lang.python')
TEST_CONF_FILE='test_search.conf'

class TestConf(unittest.TestCase):

    def testFullConf(self):
        conf = Conf("full.conf")
        self.assertEqual(conf.server, 'news.cg.shawcable.net')
        self.assertEqual(conf.context, 15)
        self.assertEqual(conf.feed_history, 50)
        self.assertEqual(conf.port, 1190)
        self.assertEqual(conf.newsrc, 'newsbiff.newsrc')

    def testDefaultConf(self):
        conf = Conf("defaults.conf")
        self.assertEqual(conf.server, 'news.cg.shawcable.net')
        self.assertEqual(conf.context, 30)
        self.assertEqual(conf.port, 119)
        self.assertEqual(conf.newsrc, 'newsbiff.newsrc')

    def testBadConf(self):
        self.assertRaises(ConfigError, Conf, 
                "bad_newsbiff.conf")

class TestSearch(unittest.TestCase):

    def setUp(self):
        """
        The NTTP server is set up and some newsgroups are created.
        """
        self.conf = Conf(TEST_CONF_FILE)
        os.environ['SNROOT'] = SNROOT
        if not os.path.isdir(SNROOT):
            os.mkdir(SNROOT)
        for group in GROUPS:
            self.createGroup(group)
        time.sleep(DELAY)
        self.server_process = subprocess.Popen(('nc6', '-l', '-p', str(SNPORT),
            '--continuous', '--exec=/usr/sbin/snntpd'), env=os.environ)
        time.sleep(DELAY)
        self.truth = True
        self.rss_test_file = "tmp/rss_test_output.xml"
        self.atom_test_files = ("tmp/atom_test_output.xml", 'author_feed.xml', 
                'thread_size_feed.xml', 'no_filter_feed.xml')

    def tearDown(self):
        """
        The NTTP server is shut down and test newsgroups are deleted.
        """
        time.sleep(DELAY)
        os.kill(self.server_process.pid, signal.SIGKILL)
        time.sleep(DELAY)
        # Ensure we are removing the spool directory
        if os.path.exists(SNROOT + "/.chain") and \
                os.path.exists(SNROOT + "/.newsgroup"):
            shutil.rmtree(SNROOT)
        for filename in (self.conf.newsrc, self.rss_test_file) \
                + self.atom_test_files: 
            if os.path.exists(filename):
                os.unlink(filename)

        time.sleep(DELAY)

    def testTruth(self):
        """
        Ensure our test environment is properly set up.
        """
        self.assertTrue(self.truth)
        # Check that the spool directory has been created
        self.assertTrue(os.path.isdir(SNROOT))
        # Check that the server process is running
        self.assertEqual(self.server_process.poll(), None) 

    def testPost(self):
        self.post((1,))
        nntpserver = nntplib.NNTP(SNHOST, SNPORT)
        (_, _, _, headers_list) = nntpserver.head('<al123@alpage.org>')
        expected_header = "Subject: Test Message 1"
        nntpserver.quit()
        self.failIf(not expected_header in headers_list)

    def testReplyTo(self):
        self.post(range(1, 3))
        engine = SearchEngine(TEST_CONF_FILE)
        self.assertEqual(len(engine.results), 1)
        # A second run should not report the message again
        engine = SearchEngine(TEST_CONF_FILE)
        self.assertEqual(len(engine.results), 0)

    def testReplysToReply(self):
        self.post(range(1, 5))
        engine = SearchEngine(TEST_CONF_FILE)
        self.assertEqual(len(engine.results), 3)

    def testReplysToReplyIn2Parts(self):
        self.post(range(1, 4))
        engine = SearchEngine(TEST_CONF_FILE)
        self.assertEqual(len(engine.results), 2)
        # A second run should not report the messages again
        engine = SearchEngine(TEST_CONF_FILE)
        self.assertEqual(len(engine.results), 0)
        self.post((4,))
        engine = SearchEngine(TEST_CONF_FILE)
        self.assertEqual(len(engine.results), 1)

    def testSubject1(self):
        self.post(range(1, 5))
        engine = SearchEngine("test_search_subject_1.conf")
        self.assertEqual(len(engine.results), 3)

    def testSubject2(self):
        self.post(range(1, 5))
        engine = SearchEngine("test_search_subject_2.conf")
        self.assertEqual(len(engine.results), 4)

    def testAuthor(self):
        self.post(range(1, 7))
        engine = SearchEngine("test_search_author.conf")
        self.assertEqual(len(engine.results), 2)

    def testAll(self):
        self.post(range(1, 9))
        engine = SearchEngine("test_search_all.conf")
        self.assertEqual(len(engine.results), 7)

    def testThreadSizeMore3(self):
        self.post(range(1, 9))
        engine = SearchEngine("test_thread_size_3.conf")
        self.assertEqual(len(engine.results), 7)

    def testThreadSizeMore4(self):
        self.post(range(1, 9))
        engine = SearchEngine("test_thread_size_4.conf")
        self.assertEqual(len(engine.results), 4)

    def testRssRendering(self):
        self.post(range(1, 9))
        engine = SearchEngine("test_thread_size_4.conf")
        RssRenderer(engine.results, self.rss_test_file).render()
        self.assert_(os.path.exists(self.rss_test_file))
        rss_content = open(self.rss_test_file).read()
        self.assert_(rss_content.find("not in the file") == -1)
        self.assert_(rss_content.find("<title>Test Message 1</title>"))

    def testAtomRendering(self):
        self.post(range(1, 9))
        engine = SearchEngine("test_thread_size_4.conf")
        renderer = AtomRenderer(self.atom_test_files[0], 'Test feed title', 
                'http://bla.org/feed/', engine.conf.feed_history)
        renderer.render(engine.results)
        self.assert_(os.path.exists(self.atom_test_files[0]))
        tree = ET.parse(self.atom_test_files[0])
        ATOM_NS = AtomRenderer.ATOM_NS
        self.assertEqual('{%s}feed' % ATOM_NS, tree.getroot().tag)
        self.assertEqual('Test feed title', tree.findtext("{%s}title" % ATOM_NS))
        self.assertEqual(4, len(tree.findall("{%s}entry" % ATOM_NS)))

        engine = SearchEngine("test_thread_size_3.conf")
        atom_content= open(self.atom_test_files[0]).read()
        self.assert_('First message of a thread of 4' in atom_content)
        renderer.render(engine.results)
        tree = ET.parse(self.atom_test_files[0])
        self.assertEqual(5, len(tree.findall("{%s}entry" % ATOM_NS)))
        atom_content = open(self.atom_test_files[0]).read()
        self.assert_(not 'First message of a thread of 4' in atom_content)
        self.assert_('Last message of a thread of 4' in atom_content)
        self.assert_('First message of a thread of 3' in atom_content)
        self.assert_('Last message of a thread of 3' in atom_content)

    def testIndividualAtomFeeds(self):
        self.post(range(1, 9))
        engine = SearchEngine("indiviual_feeds.conf")
        for atom_file in self.atom_test_files[1:]:
            self.assert_(os.path.exists(atom_file))
        ATOM_NS = AtomRenderer.ATOM_NS
        tree = ET.parse('author_feed.xml')
        self.assertEqual(5, len(tree.findall("{%s}entry" % ATOM_NS)))
        tree = ET.parse('thread_size_feed.xml')
        self.assertEqual(4, len(tree.findall("{%s}entry" % ATOM_NS)))
        tree = ET.parse('no_filter_feed.xml')
        self.assertEqual(8, len(tree.findall("{%s}entry" % ATOM_NS)))

    ###### Helper methods follow ######

    def post(self, n_list):
        nntpserver = nntplib.NNTP(SNHOST, SNPORT)
        for n in n_list:
            nntpserver.post(open('samples/msg' + str(n)))
        nntpserver.quit()
        time.sleep(DELAY)

    def createGroup(self, groupname):
        devnull = open('/dev/null', 'w')
        subprocess.call(("snnewgroup", groupname), stderr=devnull, env=os.environ)
        self.assertTrue(os.path.isdir("%s/%s" % (SNROOT, groupname)))

    def deleteGroup(self, groupname):
        subprocess.call(("sndelgroup", groupname), env=os.environ)
        self.assertFalse(os.path.isdir("%s/%s" % (SNROOT, groupname)))

if __name__ == '__main__':
    unittest.main()

#EOF
